package com.bi.service.impl;

import com.bi.dto.WideTableDataQueryDTO;
import com.bi.service.ElasticsearchService;
import com.bi.vo.WideTableDataVO;
import lombok.extern.slf4j.Slf4j;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.mapping.TypeMapping;
import co.elastic.clients.elasticsearch._types.mapping.Property;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
import co.elastic.clients.elasticsearch.indices.DeleteIndexResponse;
import co.elastic.clients.elasticsearch.indices.ExistsRequest;
import co.elastic.clients.elasticsearch.indices.PutMappingRequest;
import co.elastic.clients.elasticsearch.indices.PutMappingResponse;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch._types.query_dsl.QueryBuilders;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch._types.aggregations.StatsAggregate;
import co.elastic.clients.elasticsearch.core.DeleteByQueryResponse;
import co.elastic.clients.elasticsearch.core.ScrollResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggest;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;

/**
 * Elasticsearch服务实现类 V2 - 使用 Elasticsearch Java Client
 * 
 * @author BI Platform Team
 * @since 2.0.0
 */
@Slf4j
// @Service  // 注释掉，避免与现有实现冲突
public class ElasticsearchServiceImplV2 implements ElasticsearchService {

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Override
    public boolean updateMapping(String indexName, Map<String, Object> mapping) {
        try {
            // Convert Map<String, Object> to Map<String, Property>
            Map<String, Property> properties = new HashMap<>();
            for (Map.Entry<String, Object> entry : mapping.entrySet()) {
                @SuppressWarnings("unchecked")
                Map<String, Object> fieldMapping = (Map<String, Object>) entry.getValue();
                properties.put(entry.getKey(), buildPropertyFromMapping(fieldMapping));
            }
            
            PutMappingRequest request = PutMappingRequest.of(b -> b
                .index(indexName)
                .properties(properties)
            );
            
            PutMappingResponse response = elasticsearchClient.indices().putMapping(request);
            log.info("更新映射成功: {}", indexName);
            return response.acknowledged();
        } catch (IOException e) {
            log.error("更新映射失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean createIndex(String indexName, Map<String, Object> mapping) {
        try {
            // Convert Map<String, Object> to Map<String, Property>
            Map<String, Property> properties = new HashMap<>();
            for (Map.Entry<String, Object> entry : mapping.entrySet()) {
                @SuppressWarnings("unchecked")
                Map<String, Object> fieldMapping = (Map<String, Object>) entry.getValue();
                properties.put(entry.getKey(), buildPropertyFromMapping(fieldMapping));
            }
            
            CreateIndexRequest request = CreateIndexRequest.of(b -> b
                .index(indexName)
                .mappings(TypeMapping.of(m -> m.properties(properties)))
            );
            
            CreateIndexResponse response = elasticsearchClient.indices().create(request);
            log.info("创建索引成功: {}", indexName);
            return response.acknowledged();
        } catch (IOException e) {
            log.error("创建索引失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean deleteIndex(String indexName) {
        try {
            DeleteIndexRequest request = DeleteIndexRequest.of(b -> b.index(indexName));
            DeleteIndexResponse response = elasticsearchClient.indices().delete(request);
            log.info("删除索引成功: {}", indexName);
            return response.acknowledged();
        } catch (IOException e) {
            log.error("删除索引失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean indexExists(String indexName) {
        try {
            ExistsRequest request = ExistsRequest.of(b -> b.index(indexName));
            return elasticsearchClient.indices().exists(request).value();
        } catch (IOException e) {
            log.error("检查索引是否存在失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public boolean bulkInsert(String indexName, List<Map<String, Object>> documents) {
        if (CollectionUtils.isEmpty(documents)) {
            return true;
        }

        try {
            List<BulkOperation> operations = new ArrayList<>();
            
            for (int i = 0; i < documents.size(); i++) {
                Map<String, Object> document = documents.get(i);
                final int finalI = i;
                IndexOperation<Map<String, Object>> indexOp = IndexOperation.of(io -> io
                    .index(indexName)
                    .id(String.valueOf(finalI + System.currentTimeMillis()))
                    .document(document)
                );
                operations.add(BulkOperation.of(b -> b.index(indexOp)));
            }

            BulkRequest bulkRequest = BulkRequest.of(b -> b.operations(operations));
            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkRequest);
            
            if (bulkResponse.errors()) {
                log.error("批量插入部分失败");
                return false;
            }
            
            log.info("批量插入成功，索引: {}, 文档数量: {}", indexName, documents.size());
            return true;
        } catch (IOException e) {
            log.error("批量插入失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public WideTableDataVO queryWideTableData(String indexName, WideTableDataQueryDTO queryDTO) {
        long startTime = System.currentTimeMillis();
        
        try {
            // Build bool query
            BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
            buildQuery(boolQueryBuilder, queryDTO.getConditions());
            
            // Create search request
            SearchRequest request = SearchRequest.of(b -> b
                .index(indexName)
                .query(boolQueryBuilder.build()._toQuery())
                .from((queryDTO.getCurrent() - 1) * queryDTO.getSize())
                .size(queryDTO.getSize())
            );
            
            SearchResponse<Map> response = elasticsearchClient.search(request, Map.class);
            
            WideTableDataVO result = new WideTableDataVO();
            result.setTotal(response.hits().total() != null ? response.hits().total().value() : 0L);
            result.setCurrent(queryDTO.getCurrent());
            result.setSize(queryDTO.getSize());
            result.setQueryTime(System.currentTimeMillis() - startTime);

            // Process search results with proper type handling
            List<Map<String, Object>> records = new ArrayList<>();
            for (Hit<Map> hit : response.hits().hits()) {
                if (hit.source() != null) {
                    // Safely cast Map to Map<String, Object>
                    @SuppressWarnings("unchecked")
                    Map<String, Object> source = (Map<String, Object>) hit.source();
                    records.add(source);
                }
            }
            result.setRecords(records);

            return result;
        } catch (IOException e) {
            log.error("查询数据失败: {}", indexName, e);
            throw new RuntimeException("查询数据失败", e);
        }
    }

    private void buildQuery(BoolQuery.Builder boolQueryBuilder, List<WideTableDataQueryDTO.QueryCondition> conditions) {
        if (CollectionUtils.isEmpty(conditions)) {
            boolQueryBuilder.must(QueryBuilders.matchAll().build()._toQuery());
            return;
        }
        
        for (WideTableDataQueryDTO.QueryCondition condition : conditions) {
            Query query = buildSingleQuery(condition);
            if ("or".equals(condition.getLogic())) {
                boolQueryBuilder.should(query);
            } else {
                boolQueryBuilder.must(query);
            }
        }
    }

    private Query buildSingleQuery(WideTableDataQueryDTO.QueryCondition condition) {
        String field = condition.getField();
        String operator = condition.getOperator();
        Object value = condition.getValue();
        
        switch (operator) {
            case "eq":
                return QueryBuilders.term(t -> t.field(field).value(value.toString()));
            case "gt":
                return QueryBuilders.range(r -> r.field(field).gt(JsonData.of(value)));
            case "gte":
                return QueryBuilders.range(r -> r.field(field).gte(JsonData.of(value)));
            case "lt":
                return QueryBuilders.range(r -> r.field(field).lt(JsonData.of(value)));
            case "lte":
                return QueryBuilders.range(r -> r.field(field).lte(JsonData.of(value)));
            case "like":
                return QueryBuilders.wildcard(w -> w.field(field).value("*" + value + "*"));
            case "exists":
                return QueryBuilders.exists(e -> e.field(field));
            default:
                return QueryBuilders.term(t -> t.field(field).value(value.toString()));
        }
    }


    /**
     * Build Property from field mapping configuration
     */
    private Property buildPropertyFromMapping(Map<String, Object> fieldMapping) {
        String type = (String) fieldMapping.get("type");
        
        return Property.of(builder -> {
            if ("text".equals(type)) {
                builder.text(t -> t);
            } else if ("keyword".equals(type)) {
                builder.keyword(k -> k);
            } else if ("long".equals(type)) {
                builder.long_(l -> l);
            } else if ("integer".equals(type)) {
                builder.integer(i -> i);
            } else if ("short".equals(type)) {
                builder.short_(s -> s);
            } else if ("byte".equals(type)) {
                builder.byte_(b -> b);
            } else if ("double".equals(type)) {
                builder.double_(d -> d);
            } else if ("float".equals(type)) {
                builder.float_(f -> f);
            } else if ("date".equals(type)) {
                builder.date(d -> d);
            } else if ("boolean".equals(type)) {
                builder.boolean_(b -> b);
            } else if ("binary".equals(type)) {
                builder.binary(b -> b);
            } else if ("object".equals(type)) {
                builder.object(o -> o);
            } else if ("nested".equals(type)) {
                builder.nested(n -> n);
            } else {
                // Default to text if type is not recognized
                builder.text(t -> t);
            }
            return builder;
        });
    }

    // 其他方法的实现...
    // 由于篇幅限制，这里只展示核心方法的迁移示例

    @Override
    public Map<String, Object> aggregateQuery(String indexName, WideTableDataQueryDTO queryDTO) {
        try {
            // Build base query
            BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
            buildQuery(boolQueryBuilder, queryDTO.getConditions());
            
            // Create aggregation request
            SearchRequest request = SearchRequest.of(b -> b
                .index(indexName)
                .query(boolQueryBuilder.build()._toQuery())
                .size(0) // Only return aggregations, not documents
                .aggregations("stats", a -> a
                    .stats(s -> s.field(queryDTO.getAggregation().getFunctions().get(0).getField()))
                )
            );
            
            SearchResponse<Map> response = elasticsearchClient.search(request, Map.class);
            
            Map<String, Object> result = new HashMap<>();
            if (response.aggregations() != null && response.aggregations().containsKey("stats")) {
                StatsAggregate statsAgg = response.aggregations().get("stats").stats();
                result.put("count", statsAgg.count());
                result.put("min", statsAgg.min());
                result.put("max", statsAgg.max());
                result.put("avg", statsAgg.avg());
                result.put("sum", statsAgg.sum());
            }
            
            return result;
        } catch (IOException e) {
            log.error("聚合查询失败: {}", indexName, e);
            throw new RuntimeException("聚合查询失败", e);
        }
    }

    @Override
    public WideTableDataVO.FieldStats getFieldStats(String indexName, String fieldName) {
        try {
            SearchRequest request = SearchRequest.of(b -> b
                .index(indexName)
                .size(0)
                .aggregations("stats", a -> a
                    .stats(s -> s.field(fieldName))
                )
            );
            
            SearchResponse<Map> response = elasticsearchClient.search(request, Map.class);
            WideTableDataVO.FieldStats stats = new WideTableDataVO.FieldStats();
            
            if (response.aggregations() != null && response.aggregations().containsKey("stats")) {
                StatsAggregate statsAgg = response.aggregations().get("stats").stats();
                stats.setCount(statsAgg.count());
                stats.setMin(statsAgg.min());
                stats.setMax(statsAgg.max());
                stats.setAvg(statsAgg.avg());
                stats.setSum(statsAgg.sum());
            }
            
            return stats;
        } catch (IOException e) {
            log.error("获取字段统计失败: {} - {}", indexName, fieldName, e);
            throw new RuntimeException("获取字段统计失败", e);
        }
    }

    @Override
    public long getDocumentCount(String indexName) {
        try {
            SearchRequest request = SearchRequest.of(b -> b
                .index(indexName)
                .size(0)
                .query(QueryBuilders.matchAll().build()._toQuery())
            );
            
            SearchResponse<Map> response = elasticsearchClient.search(request, Map.class);
            return response.hits().total() != null ? response.hits().total().value() : 0L;
        } catch (IOException e) {
            log.error("获取文档数量失败: {}", indexName, e);
            throw new RuntimeException("获取文档数量失败", e);
        }
    }

    @Override
    public boolean deleteByQuery(String indexName, Map<String, Object> query) {
        try {
            // Convert Map query to Elasticsearch query
            Query esQuery = QueryBuilders.term(t -> t
                .field((String) query.get("field"))
                .value(query.get("value").toString())
            );
            
            DeleteByQueryResponse response = elasticsearchClient.deleteByQuery(d -> d
                .index(indexName)
                .query(esQuery)
            );
            
            log.info("删除查询成功: {}, 删除数量: {}", indexName, response.deleted());
            return true;
        } catch (IOException e) {
            log.error("删除查询失败: {}", indexName, e);
            return false;
        }
    }

    @Override
    public List<String> suggest(String indexName, String field, String prefix, int size) {
        try {
            SearchRequest request = SearchRequest.of(b -> b
                .index(indexName)
                .suggest(s -> s
                    .suggesters("suggestion", sug -> sug
                        .prefix(prefix)
                        .completion(c -> c
                            .field(field)
                            .size(size)
                        )
                    )
                )
            );
            
            SearchResponse<Map> response = elasticsearchClient.search(request, Map.class);
            List<String> suggestions = new ArrayList<>();
            
            if (response.suggest() != null && response.suggest().containsKey("suggestion")) {
                Suggestion<Map> suggestion = response.suggest().get("suggestion").get(0);
                if (suggestion.isCompletion()) {
                    CompletionSuggest<Map> completionSuggest = suggestion.completion();
                    for (CompletionSuggestOption option : completionSuggest.options()) {
                        suggestions.add(option.text());
                    }
                }
            }
            
            return suggestions;
        } catch (IOException e) {
            log.error("搜索建议失败: {} - {}", indexName, field, e);
            throw new RuntimeException("搜索建议失败", e);
        }
    }

    @Override
    public void exportData(String indexName, WideTableDataQueryDTO queryDTO, DataExportCallback callback) {
        try {
            // Build query
            BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
            buildQuery(boolQueryBuilder, queryDTO.getConditions());
            
            // Create scroll search for large data export
            SearchRequest request = SearchRequest.of(b -> b
                .index(indexName)
                .query(boolQueryBuilder.build()._toQuery())
                .size(1000) // Batch size
                .scroll(s -> s.time("5m"))
            );
            
            SearchResponse<Map> response = elasticsearchClient.search(request, Map.class);
            final String[] scrollIdHolder = new String[]{response.scrollId()};
            
            // Process initial batch
            processExportBatch(response, callback);
            
            // Continue scrolling through all results
            while (true) {
                ScrollResponse<Map> scrollResponse = elasticsearchClient.scroll(scrollReq -> scrollReq
                    .scrollId(scrollIdHolder[0])
                    .scroll(scroll -> scroll.time("5m"))
                , Map.class);
                
                if (scrollResponse.hits().hits().isEmpty()) {
                    break;
                }
                
                processExportBatch(scrollResponse, callback);
                scrollIdHolder[0] = scrollResponse.scrollId();
            }
            
            // Clear scroll context
            elasticsearchClient.clearScroll(c -> c.scrollId(scrollIdHolder[0]));
            
        } catch (IOException e) {
            log.error("数据导出失败: {}", indexName, e);
            throw new RuntimeException("数据导出失败", e);
        }
    }

    private void processExportBatch(SearchResponse<Map> response, DataExportCallback callback) {
        List<Map<String, Object>> batch = new ArrayList<>();
        for (Hit<Map> hit : response.hits().hits()) {
            if (hit.source() != null) {
                @SuppressWarnings("unchecked")
                Map<String, Object> source = (Map<String, Object>) hit.source();
                batch.add(source);
            }
        }
        callback.process(batch);
    }

    private void processExportBatch(ScrollResponse<Map> response, DataExportCallback callback) {
        List<Map<String, Object>> batch = new ArrayList<>();
        for (Hit<Map> hit : response.hits().hits()) {
            if (hit.source() != null) {
                @SuppressWarnings("unchecked")
                Map<String, Object> source = (Map<String, Object>) hit.source();
                batch.add(source);
            }
        }
        callback.process(batch);
    }
}